package com.xuzz.xrpc.registry;

import com.xuzz.xrpc.common.ConfigManager;
import com.xuzz.xrpc.common.PropertyConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @description:
 * @author: xuzz
 * @create: 2020-02-21 21:26
 */
public class ZookeeperRegistry extends AbstractRegistrar {
    private static final int SESSION_TIMEOUT_MS = 5000;
    private static final int SLEEP_TIME_MS = 1000;
    private static final int MAX_RETRIES = 2;

//    public ZookeeperRegistry(PropertyConfig propertyConfig)
//    {
//        super(propertyConfig);
//    }

    private Map<String, List<String>> serviceProviderMap = new HashMap<String, List<String>>();
    private CuratorFramework curatorFramework;

    @Override
    protected void init(String registerAddress){
        curatorFramework = CuratorFrameworkFactory.builder()
                .connectString(registerAddress)
                .sessionTimeoutMs(SESSION_TIMEOUT_MS)
                .retryPolicy(new ExponentialBackoffRetry(SLEEP_TIME_MS, MAX_RETRIES))
                .build();

        curatorFramework.start();

    }

    public void register(String service) {
        try {
            String servicePath = FOLDER + SEPARATOR + service;
            Stat stat = curatorFramework.checkExists().forPath(servicePath);
            if (stat == null) {
                curatorFramework.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT).forPath(servicePath);
            }
            String provider = servicePath + SEPARATOR + ConfigManager.propertyConfig.getServerAddress();

            curatorFramework.create().withMode(CreateMode.EPHEMERAL)
                    .forPath(provider);
        } catch (Exception e) {
            System.out.println(e);
        }
    }


    @Override
    public List<String> lookup(String service) {
        String path = FOLDER + SEPARATOR + service;
        try {
            List<String> provider = curatorFramework.getChildren().forPath(path);
            serviceProviderMap.put(service, provider);

            watchProvider(path);

            return serviceProviderMap.get(service);
        } catch (Exception e) {
            return null;
        }
    }

    private void watchProvider(final String path) {
        PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, path, true);
        PathChildrenCacheListener listener = new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                serviceProviderMap.put(path, curatorFramework.getChildren().forPath(path));
            }
        };
        childrenCache.getListenable().addListener(listener);
        try {
            childrenCache.start();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }
}
